package com.wsjj.yjh;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.time.Duration;

public class KeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(2000L, CheckpointingMode.EXACTLY_ONCE);
        KeyedStream<WaterSensor, String> stream = env.socketTextStream("hadoop102", 7777)
                .map(value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs()*1000L;
                            }
                        }))
                .keyBy(WaterSensor::getId);
        stream.process(new KeyedProcessFunction<String, WaterSensor, String>() {

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
//                获取当前数据的key
                String currentKey = ctx.getCurrentKey();
                System.out.println("key的信息："+currentKey);
//                1.定时器注册
                TimerService timerService = ctx.timerService();

//                注册定时器： 处理时间
//                timerService.registerProcessingTimeTimer();
//                删除定时器： 事件时间
//                timerService.deleteEventTimeTimer();
//                删除定时器：  处理时间
//                timerService.deleteProcessingTimeTimer();
//                获取当前处理时间：系统时间
                timerService.currentProcessingTime();
//                获取当前的  watermark
                timerService.currentWatermark();
//
                Long timestamp = ctx.timestamp();   //数据中提取出来的事件时间
                timerService.registerEventTimeTimer(5000L);
                System.out.println("当前key=" + currentKey + ",当前时间=" + timestamp + ",注册了一个5s的定时器");

            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                String currentKey = ctx.getCurrentKey();
                System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");
            }
        });

        env.execute();


    }
}
